package com.open.source.kafka;

import com.open.source.MqConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.Headers;
import org.springframework.kafka.listener.ConsumerAwareRecordInterceptor;

/**
 * @author ZonLen since on 2022/8/27 下午11:27
 */
@Slf4j
public class TraceRecordInterceptor implements ConsumerAwareRecordInterceptor<String, String> {

//  @Autowired(required = false)
//  private ICasCorpAdaptor casCorpAdaptor;

  @Override
  public ConsumerRecord<String, String> intercept(ConsumerRecord<String, String> record,
      Consumer<String, String> consumer) {
    final Headers headers = record.headers();
//    final String requestId = new String(headers.lastHeader(Constants.REQUEST_ID_KEY).value());
//    MDC.put(Constants.REQUEST_ID_KEY, requestId);
//    final Header header = headers.lastHeader(Constants.X_TENANT_ID);
//    if (null != header) {
//      final String corpId = new String(header.value());
//      if (StringUtils.isNotBlank(corpId)) {
//        CasCorp casCorp = casCorpAdaptor.getCorpByid(corpId);
//        if (casCorp == null) {
//          casCorp = new CasCorp();
//          casCorp.setCorpId(corpId);
//        }
//        InvocationInfoUtils.addEntity(CasCorp.class, casCorp);
//      }
//    }
    final Header header = headers.lastHeader(MqConstant.PRODUCER_HOST_IP);
    String hostIp = "";
    if (null != header) {
      hostIp = new String(header.value());
    }
    String appName = "";
    if (null != header) {
      appName =
          new String(headers.lastHeader(MqConstant.PRODUCER_APPLICATION_NAME).value());
    }
    log.info(
        "TraceRecordInterceptor PRODUCER_HOST_IP:{}, PRODUCER_APPLICATION_NAME:{}, consumer-group-id:{}, topic:{}, partition:{}, value:{}",
        hostIp, appName, consumer.groupMetadata().groupId(), record.topic(), record.partition(),
        new String(record.value()));

    return record;
  }

  /**
   * 监听器正常结束
   */
  @Override
  public void success(ConsumerRecord<String, String> record, Consumer<String, String> consumer) {
//    InvocationInfoUtils.cleanAll();
//    MDC.clear();
  }

  /**
   * 监听器异常结束
   */
  @Override
  public void failure(ConsumerRecord<String, String> record, Exception exception,
      Consumer<String, String> consumer) {
    final Headers headers = record.headers();
    final String hostIp =
        new String(headers.lastHeader(MqConstant.PRODUCER_HOST_IP).value());
    final String appName =
        new String(headers.lastHeader(MqConstant.PRODUCER_APPLICATION_NAME).value());
    log.error(
        "Consumer主题||分组<{}||{}>消费异常,消息来源=》PRODUCER_HOST_IP:{}, PRODUCER_APPLICATION_NAME:{}, \n数据=> {} \n异常=> ",
        record.topic(), consumer.groupMetadata().groupId(), hostIp, appName, new String(record.value()),
        exception);
//    InvocationInfoUtils.cleanAll();
//    MDC.clear();
  }
}
